package com.fc.test.compare;

import com.fc.test.common.domain.BaseObject;
import com.fc.test.compare.define.CompareInterface;
import com.fc.test.compare.define.CompareResult;
import com.github.rholder.retry.Attempt;
import com.github.rholder.retry.RetryListener;
import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import com.google.common.collect.Maps;
import org.apache.commons.dbutils.QueryRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by Liaopan on 2019/7/17.
 */
public class CompareColumnsReturnMap implements CompareInterface {
    private static final Logger LOG = LoggerFactory.getLogger(CompareColumnsReturnMap.class);

    DataSource source;
    DataSource target;

    private Class<BaseObject> sourceClass;
    private Class<BaseObject> targetClass;

    @Override
    public void init(DataSource sds, DataSource tds, Class sourceClass, Class targetClass) {
        this.source = sds;
        this.target = tds;
        this.sourceClass = sourceClass;
        this.targetClass = targetClass;
    }

    @Override
    public CompareResult compare(final String sourceQuery, final String targetQuery) throws Exception {
        QueryRunner sourceR = new QueryRunner(source, true);
        QueryRunner targetR = new QueryRunner(target, true);

        LOG.debug("====查询sql====");
        LOG.debug("source:" + sourceQuery);
        LOG.debug("target:" + targetQuery);
        LOG.debug("====end====");
        CompareResultMap resultList = null;

        final CompareResultMap.Builder builder = new CompareResultMap.Builder();
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        final CountDownLatch latch = new CountDownLatch(2);

        Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder()
                .retryIfExceptionOfType(SQLException.class)
                .withStopStrategy(StopStrategies.stopAfterAttempt(3))
                .withRetryListener(new RetryListener() {
                    @Override
                    public <V> void onRetry(Attempt<V> attempt) {
                        LOG.error(attempt.getExceptionCause().getMessage());
                        LOG.info("准备第"+attempt.getAttemptNumber()+"次重试");
                    }
                }).withWaitStrategy(WaitStrategies.fixedWait(5000, TimeUnit.SECONDS))
                .build();

        cachedThreadPool.submit(() -> {
            try {
                retryer.call(() -> {
                    builder.left(sourceR.query(sourceQuery, new MapClassResultHandler(sourceClass)));
                    return true;
                });
            } catch (Exception e) {
                LOG.error("原表查询出错！", e);
                builder.left(Maps.newHashMap());
                builder.lMessage(e.getMessage());
            } finally {
                latch.countDown();
            }
        });

        cachedThreadPool.submit(() -> {
            try {
                retryer.call(() -> {
                    builder.right(targetR.query(targetQuery, new MapClassResultHandler(targetClass)));
                    return true;
                });
            } catch (Exception e) {
                LOG.error("目标表查询出错！", e);
                builder.right(Maps.newHashMap());
                builder.rMessage(e.getMessage());
            } finally {
                latch.countDown();
            }
        });

        latch.await();
        // 比对数据
        resultList = builder.builder();

        return resultList;
    }
}
